package net.bwie.realtime.vehicle.function;

import net.bwie.realtime.vehicle.utils.PGIllegalVehicleLoader;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.util.Set;

/**
 * 过滤函数：保留违法次数超30次的车辆数据
 * @author ASUS
 */
public class IllegalVehicleFilter extends RichFilterFunction<String> {
    private final PGIllegalVehicleLoader loader;
    private Set<String> illegalCars;

    public IllegalVehicleFilter(PGIllegalVehicleLoader loader) {
        this.loader = loader;
    }

    @Override
    public void open(Configuration parameters) {
        // 预加载违法车辆数据（仅在算子初始化时执行一次）
        illegalCars = loader.loadIllegalCars();
    }

    @Override
    public boolean filter(String vehicleData) {
        // 从流数据中解析出车牌号（需根据实际JSON格式调整）
        String carId = parseCarId(vehicleData);
        // 保留在违法车辆列表中的数据
        return carId != null && illegalCars.contains(carId);
    }

    /**
     * 解析流数据中的车牌号
     * 实际场景需根据JSON结构修改（建议使用Jackson/Gson）
     */
    private String parseCarId(String vehicleData) {
        try {
            // 示例：假设数据格式为 {"car":"皖J73691", ...}
            // 使用JSON解析库提取car字段（此处简化处理，实际需替换为真实解析逻辑）
            if (vehicleData.contains("\"car\":\"")) {
                String carPart = vehicleData.split("\"car\":\"")[1];
                return carPart.split("\"")[0];
            }
            return null;
        } catch (Exception e) {
            System.err.println("解析车牌号失败：" + vehicleData);
            return null;
        }
    }
}
